package com.lwq.fast.log.fastlogserver.redis.listener;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.lwq.fast.log.fastlogcore.constant.Constants;
import com.lwq.fast.log.fastlogcore.entity.Message;
import com.lwq.fast.log.fastlogcore.util.ThreadPoolUtil;
import com.lwq.fast.log.fastlogserver.es.service.ElasticSearchService;
import com.lwq.fast.log.fastlogserver.redis.config.RedisConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;


/**
 * 消息监听器，监听redis消息
 * @author 刘文强
 */
@Configuration
@ConditionalOnProperty(name = "active.client", havingValue = "redis")
@Slf4j
public class RedisConsumerListener {


    @Autowired
    private JedisPool jedisPool;
    @Value("${redis.sleep.timeout}")
    private int timeOut;

    @Autowired
    private ElasticSearchService elasticSearchService;

    @Bean
    public void onMessage(){
        ThreadPoolExecutor threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();
        threadPoolExecutor.execute(() ->{
            while (true){
                Jedis jedis = jedisPool.getResource();
                try {
                    List<String> strMsgList = jedis.brpop(timeOut, Constants.REDIS_LOG_KEY);
                    if (CollectionUtil.isNotEmpty(strMsgList)){
                        strMsgList.forEach(msg ->{
                            Message message = null;
                            try{
                                message = JSON.parseObject(msg, Message.class);
                                if (ObjectUtil.isNotNull(message)){
                                    // 如果不能转换成 Message 类型，就抛弃
                                    elasticSearchService.logMessage2Es(msg);
                                }
                            }catch (Exception formatException){
                            }
                        });
                    }
                }catch (Exception e){
                    log.error("RedisConsumerListener onMessage error ", e);
                }finally {
                    if (null != jedis){
                        jedis.close();
                    }
                }
            }
        });

    }

}
